package com.zxb.core.socket;

import cn.hutool.core.collection.CollUtil;
import cn.hutool.core.util.StrUtil;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;

import javax.websocket.*;
import javax.websocket.server.PathParam;
import javax.websocket.server.ServerEndpoint;
import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ConcurrentHashMap;

/**
 * @author Zxb
 */
@SuppressWarnings("unused")
@ServerEndpoint("/ws/{clientId}")
@Component
@Slf4j
public class WebSocketServer {
    /**
     * 与某个客户端的连接会话，需要通过它来给客户端发送数据
     */
    private Session session;

    /**
     * session集合,存放对应的session
     */
    public static ConcurrentHashMap<String, List<Session>> clientPool = new ConcurrentHashMap<>();

    /**
     * 建立WebSocket连接
     *
     * @param session  会话
     * @param clientId 客户端ID
     */
    @OnOpen
    public void onOpen(Session session, @PathParam(value = "clientId") String clientId) {
        log.info("开始建立链接，客户端ID：{}，sessionId：{}", clientId, session.getId());
        List<Session> list = clientPool.get(clientId);
        if (list == null) {
            list = new ArrayList<>();
            list.add(session);
            clientPool.put(clientId, list);
        } else {
            list.add(session);
        }
        // 建立连接
        this.session = session;
        log.info("成功建立链接，当前客户端会话数：{}，总客户端数：{}", list.size(), clientPool.size());
    }

    /**
     * 发生错误
     *
     * @param throwable e
     */
    @OnError
    public void onError(Throwable throwable) {
        log.error("发生错误");
        throwable.printStackTrace();
    }

    /**
     * 连接关闭
     */
    @OnClose
    public void onClose(Session session, CloseReason closeReason) {
        for (String str : clientPool.keySet()) {
            List<Session> oldList = clientPool.get(str);
            List<Session> newList = new ArrayList<>();
            oldList.stream().forEach(t -> {
                if (t.isOpen()) {
                    newList.add(t);
                }
            });
            clientPool.put(str, newList);
        }
        log.info("链接关闭:" +
                session.getId());
    }

    /**
     * 接收客户端消息
     *
     * @param message 接收的消息
     */
    @OnMessage
    public void onMessage(String message) throws IOException {
        if (StrUtil.isNotBlank(message)
                && "heart".equals(message)) {
            session.getBasicRemote().sendText("heart copy");
        }
        //log.info("收到客户端发来的消息：{}", message);
        //session.getBasicRemote().sendText(message);
    }

    /**
     * 推送消息到指定客户端
     *
     * @param clientId 客户端ID
     * @param message  发送的消息
     */
    public static void sendMsg(String clientId, String message) {
        synchronized (clientId) {

            //log.info("向客户端【" + clientId + "】推送内容：" + message);
            List<Session> sessionList = clientPool.get(clientId);
            if (CollUtil.isEmpty(sessionList)) {
                return;
            }
            try {
                for (Session session : sessionList) {
                    if (session.isOpen()) {
                        session.getBasicRemote().sendText(message);
                    }
                }
            } catch (IOException e) {
                log.error("推送消息到客户端【" + clientId + "】时发生错误：" + e.getMessage(), e);
            }
        }
    }

    /**
     * 群发消息
     *
     * @param message 发送的消息
     */
    public static void sendMsg(String message) {
        log.info("群发消息：{}", message);
        for (String clientId : clientPool.keySet()) {
            sendMsg(clientId, message);
        }
    }
}
